package com.doit.spark.day06

import com.doit.spark.day01.utils.SparkUtil
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD

/**
 * @DATE 2022/1/9/15:28
 * @Author MDK
 * @Version 2021.2.2
 *
 *  统计每个省份的数据访问量
 *  用户的日志数据  数据访问量较大
 *  每个task上都会进行规则匹配 每个task上有完整的数据
 *  [ip规则数据  地理位置数据  手机地理位置数据]
 *
 * */
object Demo01 {
  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSc
    val ipRuleDataRDD = sc.textFile("data/ip/ip.txt")
//    ipRuleDataRDD.foreach(println)

//    println("---------------------------------------")
    //封装数据
    val ipRuleBeans: RDD[IpRuleBean] = ipRuleDataRDD.map(line => {
      val arr: Array[String] = line.split("\\|")
      val bean = IpRuleBean(arr(2).toLong,
        arr(3).toLong,
        arr(4),
        arr(5),
        arr(6),
        arr(7),
        arr(8))
      bean
    })
//    ipRuleBeans.foreach(println)

    //收集到Driver端  提交数据,将数据广播出去
    val arr: Array[IpRuleBean] = ipRuleBeans.collect()
    val bc: Broadcast[Array[IpRuleBean]] = sc.broadcast(arr)


    //加载日志数据
    val logRDD: RDD[String] = sc.textFile("data/ip/ipaccess.log")

    val res2: RDD[String] = logRDD.map(line => {
      var res = ""
      //获取日志数据的ip地址
      val ip = line.split("\\|")(1)
      //根据ip获取地理位置信息
      val value: Array[IpRuleBean] = bc.value
      //二分法查找
      val index = IpUtils.binarySearch(value, ip)
      //判断给定的地址是否存在
      if (index >= 0) {
        val bean = value(index)
        res = bean.province + "-" + bean.city + "-" + line
      } else {
        res = "未知省份" + "-" + "未知城市" + "-" + line
      }
      res
    })

    res2.saveAsTextFile("data/ip/ip_out")
  }
}
